package com.bg.zk.config;

import com.bg.constant.Constants;
import com.bg.utisl.PropertieConfigurationUtil;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

/**
 * @Auther: zhengshunzhi
 * @Date: 2018/12/20 15:52
 * @Description: 原生zk的api
 */
public class ZkConfigHelper implements ZkConfig, Watcher {

    //操作的节点
    private String zkPath;

    //同步
    private static CountDownLatch connectedSemaphore = new CountDownLatch(1);

    // 创建 ZooKeeper 实例
    private ZooKeeper zk = null;

    public ZkConfigHelper(String servers, int timeout, String zkPath) throws Exception {
        this.zkPath = zkPath;
        zk = new ZooKeeper(servers, timeout, this);
        connectedSemaphore.await();
    }


    public void process(WatchedEvent event) {
        Event.EventType type = event.getType();
        switch (type) {
            case NodeCreated:
                System.out.println("新建节点:" + event.getPath());
            case NodeDeleted:
                System.out.println("删除节点:" + event.getPath());
            case NodeDataChanged:
                System.out.println("修改节点:" + event.getPath());
            case NodeChildrenChanged:
                System.out.println("子节点:" + event);
        }

        if (Event.KeeperState.SyncConnected == event.getState()) {
            //释放阻塞
            connectedSemaphore.countDown();
        } else if (event.getState().equals(Event.KeeperState.Disconnected)) {
            // 这时收到断开连接的消息，这里其实无能为力，因为这时已经和ZK断开连接了，只能等ZK再次开启了
            System.out.println("zk Disconnected");
        } else if (event.getState().equals(Event.KeeperState.Expired)) {
            // 这时收到这个信息，表示，ZK已经重新连接上了，但是会话丢失了，这时需要重新建立会话。
            System.out.println("zk Expired");
            //这里可以进行zk重新连接操作
            //do Some thing....
        } else if (event.getState().equals(Event.KeeperState.AuthFailed)) {
            System.out.println("zk AuthFailed");
        } else {
            System.out.println("unkonke...." + event.getType());
        }

        try {
            //监听器只会触发一次，需要我们重复监听
            zk.getChildren(zkPath, true);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 功能描述: 增加节点
     *
     * @param: key:节点
     * @return: value:数据
     * @auther: bonree
     * @date: 2018/12/20 16:12
     * 节点：key , 数据： value ，权限： OPEN_ACL_UNSAFE ，节点类型： Persistent
     */
    @Override
    public String add(String key, String value) throws KeeperException, InterruptedException, UnsupportedEncodingException {
        if (!exsitsKey(zkPath)) {
            zk.create(zkPath, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
        key = zkPath + "/" + key;
        return zk.create(key, value.getBytes("utf-8"), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }

    /**
     * 功能描述: 修改节点
     *
     * @param: key
     * @param: value
     * @return:
     * @auther: bonree
     * @date: 2018/12/20 16:26
     */
    @Override
    public boolean update(String key, String value) throws KeeperException, InterruptedException {
        Stat stat = zk.setData(zkPath + "/" + key, value.getBytes(), -1);
        return stat == null ? false : true;
    }

    /**
     * 功能描述: 删除节点
     *
     * @param: key
     * @return:
     * @auther: bonree
     * @date: 2018/12/20 16:27
     */
    @Override
    public void delete(String key) throws KeeperException, InterruptedException {
        zk.delete(zkPath + "/" + key, -1);
    }

    /**
     * 功能描述: 通过key获取value
     *
     * @param: key
     * @return: value
     * @auther: bonree
     * @date: 2018/12/20 16:18
     */
    @Override
    public String getKey(String key) throws KeeperException, InterruptedException {
        return new String(zk.getData(key, false, null));
    }

    /**
     * 功能描述: 查看节点是否存在
     *
     * @param: key
     * @return:
     * @auther: bonree
     * @date: 2018/12/20 16:31
     */
    @Override
    public boolean exsitsKey(String key) throws KeeperException, InterruptedException {
        return zk.exists(key, false) == null ? false : true;
    }


    /**
     * 功能描述:
     *
     * @param:
     * @return:
     * @auther: bonree
     * @date: 2018/12/20 17:44
     */
    @Override
    public Map<String, String> getAll(String rootPath) throws KeeperException, InterruptedException {
        Map<String, String> res = new ConcurrentHashMap<>();
        List<String> children = zk.getChildren(rootPath, false);
        children.forEach(key -> {
            try {
                String value = getKey(rootPath + "/".concat(key));
                res.put(key, value);
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        return res;
    }

    private void ZKClose() throws InterruptedException {
        zk.close();
    }
}
